package com.haibara.codesandbox.kafka.processor;

import cn.hutool.json.JSONUtil;
import com.haibara.codesandbox.constant.KafkaConstant;
import com.haibara.codesandbox.model.ExecuteCodeRespond;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/**
 * 用于发送消息至Kafka的类
 *
 * @Author: mengyu
 * @Date: 2025/2/10
 */
@Component
@Slf4j
public class Processor {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 发送消息
     *
     * @param topic 主题
     * @param msg   信息
     */
    public void sendMsg(String topic, String msg) {
        CompletableFuture<SendResult<String, String>> send = this.kafkaTemplate.send(topic, msg);
        send.thenAcceptAsync(result -> {
            log.info("topic:{} ,msg: {} ,发送成功!", topic, msg);
        }).exceptionally(ex -> {
            log.error("topic:{} ,msg: {} ,发送失败!", topic, msg);
            log.error("exception: ", ex);
            return null;
        });
    }
}
